"""
author: liuxu
date: 2024.12.3
description: 此模块用于测试hipprof的tracing功能
"""
import math
import random
import allure
import pytest
import re

from common.logger_ctrl import mylogger
from common.ssh_command import exec_cmd
from base_public.testbase import BasePublic



@allure.feature('hipprof帮助信息')
@pytest.mark.low
class TestHipProfHelp(BasePublic):
    @pytest.mark.parametrize("paras", ["-h", "--help"])
    def test_hipprof_help(self, paras, test_env):
        """
        测试hipprof帮助信息
        :return:
        """
        ret, output = exec_cmd("hipprof",  paras)
        assert ret is True, "Failed to run hipprof, test environment needs to be checked!!!"
        assert self.check_help_info(output), f"'hipprof {paras}' output does not meet expectations"

    def test_hipprof_invalid_para(self, test_env):
        """
        测试hipprof+无效参数
        :return:
        """
        ret, output = exec_cmd("hipprof",  "--abc", "./test")
        assert ret is True
        assert self.check_help_info(output)

    def test_hipprof_invalid_exe(self, test_env):
        """
        测试hipprof+无效可执行文件
        :return:
        """
        ret, output = exec_cmd("hipprof",  "./nonexistent_file")
        assert ("No such file or directory" in output
                or "The executable program './nonexistent_file' does not exist or is not executable!" in output)


@allure.feature('hip tracing指令')
@pytest.mark.medium
class TestHipProfTrace(BasePublic):
    """测试hip trace功能"""
    # matrix用例api调用顺序
    matrix_call_seq = ['hipGetDeviceProperties', 'hipMalloc', 'hipMalloc', 'hipMemcpy', 'hipLaunchKernel', 'hipMemcpy', 'hipFree', 'hipFree']
    @pytest.mark.smoke
    @pytest.mark.high
    def test_hipprof_default(self, test_env):
        """
        测试hipprof不加参数（默认开启--hip-trace）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --trace-args {exe_path}", cwd=test_env)
        assert ret is True
        # assert self.check_trace_output(output, test_env) is True
        # 检查调用的api次数是否符合预期
        expect = {"hip_api_calls": 10, "kernel_calls": 1, "kernel_paras": "(1,256,256),(1,4,4)"}
        trace_d = self.get_hip_trace_info_from_output(output)
        assert expect["hip_api_calls"] == trace_d["api_total_calls"]  \
                and expect["kernel_calls"] == trace_d["kernel_total_calls"]  \
                and expect["kernel_paras"] == trace_d["kernel_stat"]["matrixTranspose(float*, float*, int)"]["paras"]

    def test_trace(self, test_env):
        """
        测试--hip-trace命令
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

    @pytest.mark.high
    def test_barrier(self, test_env):
        """
        测试--barrier命令（在 JSON文件中增加event和隐式同步的信息）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --barrier {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True
        # 检查json文件是否存在BARRIER信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        trace_list = self.parse_prof_json(json_path=json_path)
        barrier_count = 0 # 初始化json中获取到的barrier行个数
        for info_dict in trace_list:
            for k, v in info_dict.items():
                if k == "name" and v =="BARRIER":
                    barrier_count += 1
        mylogger.info(f"barrier count: {barrier_count}")
        assert barrier_count >= 1, f"No barrier information is queried in the json({json_path}) file"

    @pytest.mark.high
    def test_trace_segment_size_min(self, test_env):
        """
        通过--segment-size设置条目大小控制json文件大小(最小支持5000)
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/hipDispatchLatency/hipDispatchLatency.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --segment-size 5000 {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True
        # 检测生成的json文件数目是否大于2
        json_list = self.get_all_prof_files_info(output, test_env)
        assert len(json_list) > 1

    @pytest.mark.low
    def test_trace_segment_size_500000(self, test_env):
        """
        验证--segment-size设置条目大小是否默认是500000
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/hipDispatchLatency/hipDispatchLatency.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace {exe_path}", cwd=test_env, timeout=60)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 获取josn数目
        trace_d = self.get_hip_trace_info_from_output(output)
        json_num = trace_d["dump_json_num"]
        expect_json_num = math.ceil(json_num / 500000)

        # 检测生成的json文件只有1个
        json_list = self.get_all_prof_files_info(output, test_env)["json_list"]
        assert len(json_list) == expect_json_num

    def test_trace_segment_size_invalid(self, test_env):
        """
        通过--segment-size设置条目大小为无效值
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --segment-size 4999 {exe_path}", cwd=test_env)
        # assert ret is False
        assert self.check_help_info(output) is True  # 参数无效，打印help内容

    @pytest.mark.high
    @pytest.mark.skip(reason="This parameter is no longer supported after dtk-25.04.2")
    def test_trace_hid_args(self, test_env):
        """
        通过--hid-args,只统计函数名称，不统计函数参数
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --hid-args {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 检查json文件的api和kernel函数是否包含参数信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        trace_list = self.parse_prof_json(json_path=json_path)
        for info_dict in trace_list:
            if "name" in info_dict and "args" in info_dict:
                assert info_dict["args"].get("args", "") == "", f'{info_dict["name"]}函数的参数不为空'

    @pytest.mark.high
    def test_trace_trace_args(self, test_env):
        """
        通过--trace-args,开启函数参数跟踪
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --trace-args {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 检查json文件的api和kernel函数是否包含参数信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        trace_list = self.parse_prof_json(json_path=json_path)
        for info_dict in trace_list:
            if info_dict.get("ph") == "X":
                if "name" in info_dict and "args" in info_dict and "args" in info_dict["args"]:
                    assert info_dict["args"].get("args", "") != "", f'{info_dict["name"]}函数的参数未获取到'

    @pytest.mark.high
    def test_trace_stats(self, test_env):
        """
        --stats: 不输出可视化json文件
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --stats {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True
        # 检测输出json文件个数为0
        json_list = self.get_all_prof_files_info(output, test_env)["json_list"]
        assert len(json_list) == 0

    @pytest.mark.high
    def test_print_trace(self, test_env):
        """
        --print-trace: 终端打印性能分析的HIPAPI、COPY等详细数据
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --print-trace {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True
        # 检测终端是否按顺序打印api调用信息
        api_call_sequence = "hipGetDeviceProperties,hipMalloc,hipMalloc,hipMemcpy,hipLaunchKernel,hipMemcpy,hipFree,hipFree".split(",")
        copy_sequence = ["H2D", "D2H"]
        for line in output.splitlines():
            if line.startswith("HIPAPI"):
                if api_call_sequence[0] in line:
                    api_call_sequence.pop(0)
            elif line.startswith("COPY"):
                if f"kind:{copy_sequence[0]}" in line:
                    copy_sequence.pop(0)
        assert len(api_call_sequence) == len(copy_sequence) == 0

    def test_print_trace_and_hid_args(self, test_env):
        """
        --print-trace --hid-args: 终端打印不含有 args 信息的性能分析信息
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        # ret, output = exec_cmd(f"hipprof --hip-trace --print-trace --hid-args {exe_path}", cwd=test_env)
        ret, output = exec_cmd(f"hipprof --hip-trace --print-trace {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 检查json文件的api和kernel函数是否包含参数信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        trace_list = self.parse_prof_json(json_path=json_path)
        for info_dict in trace_list:
            if "name" in info_dict and "args" in info_dict:
                assert info_dict["args"].get("args", "") == "", f'{info_dict["name"]}函数的参数不为空'

        # 检查终端输出无函数入参信息
        for line in output.splitlines():
            assert not line.strip().startswith("args:"), "The terminal print does not hide the function parameters"

    @pytest.mark.high
    def test_trace_one_device(self, test_env, dcu_info):
        """
        --devices 0: 获取单张卡的性能数据
        :return:
        """
        exe_path = self.compile_source_code(src_file_name='multiproc_case/multithreading_multidevice.cpp')
        avail_device_ids = dcu_info["devices_id"]
        test_device_id = avail_device_ids[-1] # 测试序号最大的卡
        ret, output = exec_cmd(f"hipprof --hip-trace --devices {test_device_id} {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        expected_calls = 1  # 测例每张卡对应一个线程，只调用1次核函数
        actual_calls = self.get_hip_trace_info_from_output(output)["kernel_total_calls"]
        assert actual_calls == expected_calls, f"Expected {expected_calls} kernel calls, but found {actual_calls}"
        # 检查json文件,只有指定卡的信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        pattern = re.compile(r'\[3] Compute on Device (\d+)')
        with open(json_path, 'r', encoding='utf-8') as file:
            matches = pattern.findall(file.read())
        mylogger.info(f"jsonfile matches device num: {matches}")
        assert len(matches) == 1 and int(matches[0]) == test_device_id

    @pytest.mark.high
    def test_trace_multi_device(self, test_env, dcu_info):
        """
        --devices 0,2: 获取多张卡的性能测试数据
        :return:
        """
        cur_device_num = dcu_info.get("count", 0)
        if cur_device_num < 3:
            pytest.skip(f"The current number of cards is {cur_device_num}, which does not meet the test conditions")

        # 随机测试2张卡
        random_did_l =  random.sample(dcu_info.get("devices_id"), 2)
        test_device = ",".join([str(i) for i in random_did_l])

        exe_path = self.compile_source_code(src_file_name='multiproc_case/multithreading_multidevice.cpp')
        ret, output = exec_cmd(f"hipprof --hip-trace --devices {test_device} {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        expected_calls = 2  # 测例每张卡对应一个线程，每个线程只调用1次核函数
        actual_calls = self.get_hip_trace_info_from_output(output)["kernel_total_calls"]
        assert actual_calls == expected_calls, f"Expected {expected_calls} kernel calls, but found {actual_calls}"

        # 检测json文件,设备端tracing数据只有指定卡的信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        pattern_memcpy = re.compile(r'\[2] Memory Copy on Device (\d+)')
        pattern_kernel = re.compile(r'\[3] Compute on Device (\d+)')
        with open(json_path, 'r', encoding='utf-8') as file:
            file_info_str = file.read()
            matches_2 = pattern_memcpy.findall(file_info_str)
            matches_3 = pattern_kernel.findall(file_info_str)
        mylogger.info(f"jsonfile matches memcpy device: {matches_2}, kernel device: {matches_3}")
        assert len(matches_2) == 2 and [i in test_device for i in matches_2]
        assert len(matches_3) == 2 and [i in test_device for i in matches_3]

    @pytest.mark.high
    def test_trace_all_device(self, test_env, dcu_info):
        """
        --devices all: 获取所有卡的性能测试数据
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="multiproc_case/multithreading_multidevice.cpp")

        ret, output = exec_cmd(f"hipprof --hip-trace {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        expect_avail_device_ids = dcu_info["devices_id"]
        expect_dcu_cnt = len(expect_avail_device_ids)
        actual_calls = self.get_hip_trace_info_from_output(output)["kernel_total_calls"]
        assert actual_calls == expect_dcu_cnt, f"Expected {expect_dcu_cnt} kernel calls, but found {actual_calls}"
        # 检查json文件,只有指定卡的信息
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        pattern = re.compile(r'\[3] Compute on Device (\d+)')
        with open(json_path, 'r', encoding='utf-8') as file:
            matches = pattern.findall(file.read())
        mylogger.info(f"jsonfile matches device num: {matches}")
        assert set([int(match) for match in matches]) == set(expect_avail_device_ids)

    @pytest.mark.high
    def test_mask_trace_one_api(self, test_env):
        """
        --mask-trace: 过滤需要分析的HIP API，过滤单个api
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        # 先添加配置文件
        cfg_name = "filter_config.txt"
        api_name = "hipMalloc"
        with open(cfg_name, 'w', encoding='utf-8') as file:
            file.write(api_name)
        ret, output = exec_cmd(f"hipprof --hip-trace --mask-trace {cfg_name} {exe_path}", cwd=test_env)
        assert ret is True

        # 检测生成的api名称和调用次数是否正确
        print_api_stat_d = self.get_hip_trace_info_from_output(output)["api_stat"]
        assert len(print_api_stat_d) == 1 and print_api_stat_d.get(api_name)["calls"] == 2

    def test_mask_trace_multi_api(self, test_env):
        """
        --mask-trace: 过滤需要分析的HIP API，过滤多个api
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        # 添加配置文件
        cfg_name = "filter_config.txt"
        api_name = "hipMalloc,hipMemcpy"
        with open(cfg_name, 'w', encoding='utf-8') as file:
            file.write(api_name)
        ret, output = exec_cmd(f"hipprof --hip-trace --mask-trace {cfg_name} {exe_path}", cwd=test_env)
        assert ret is True

        # 检测生成的api名称和调用次数是否正确
        print_api_stat_d = self.get_hip_trace_info_from_output(output)["api_stat"]
        assert len(print_api_stat_d) == 2
        for api,api_d in print_api_stat_d.items():
            assert api in api_name and api_d["calls"] == 2  # hipMalloc,hipMemcpy均调用了2次

    @pytest.mark.low
    @pytest.mark.exception
    def test_mask_trace_noexist_cfg(self, test_env):
        """
        --mask-trace: 输入的配置文件不存在
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        cfg_name = "not_exist.txt"
        ret, output = exec_cmd(f"hipprof --hip-trace --mask-trace {cfg_name} {exe_path}", cwd=test_env)
        assert ret is False or self.check_trace_data_is_empty(output) is True

    @pytest.mark.low
    # @pytest.mark.xfail(raises=AssertionError, reason="用例预期失败")
    def test_mask_trace_invalid_cfg(self, test_env):
        """
        --mask-trace: 输入的配置文件格式不对，使用中文逗号
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        # 添加配置文件
        cfg_name = "filter_config.txt"
        api_name = "hipMalloc，hipMemcpy"
        with open(cfg_name, 'w', encoding='utf-8') as file:
            file.write(api_name)
            exec_cmd(f"cat {cfg_name}", cwd=test_env)
        ret, output = exec_cmd(f"hipprof --hip-trace --mask-trace {cfg_name} {exe_path}", cwd=test_env, timeout=30)
        assert ret is False or self.check_trace_data_is_empty(output)

    @pytest.mark.high
    @pytest.mark.smoke
    def test_kernel_stack(self, test_env):
        """
        --kernel-stack: 导出核函数调用堆栈信息，输出文件<result_pid>.stack.csv
        :param test_env:
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="multiproc_case/multiprocess_multithread_multidevice_test.cpp")

        # 测试导出kernel调用堆栈命令，此命令会在hip-trace基础上，额外生成<result_pid>.stack.csv
        ret, output = exec_cmd(f"hipprof --kernel-stack {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output)

        # 判断生成的kernel调用堆栈个数是否正确
        kernel_cnt = output.count("Finishing gpuWorker on device")  # 获取程序kennel执行次数
        stack_csv = self.get_all_prof_files_info(output).get("kernel_stack_csv").get("abspath")
        stack_info_d = self.get_kernel_stack_info_from_csv(stack_csv)
        assert len(stack_info_d) == kernel_cnt  # 判断kernel调用次数
        assert len(set([len(stack_info_d) for stack_info_d in stack_info_d.values()])) == 1  # 每个kernel的调用栈层级应相同
        #TODO bw每个kernel调用栈有差异
        t="""
        E       assert 2 == 1
E        +  where 2 = len({5, 6})
E        +    where {5, 6} = set([5, 5, 6, 6, 6, 6, ...])

        """

    def test_trace_off(self, test_env):
        """
        --trace-off: 在启动程序时禁用hipprof性能跟踪
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --trace-off {exe_path}", cwd=test_env)
        # assert ret is True
        # assert "No hip trace info" in output or "hip" not in output # 校验没有trace数据
        assert self.check_trace_data_is_empty(output)  # 检查无trace数据生成

    @pytest.mark.high
    def test_index_range(self, test_env):
        """
        --index-range: 指定HIP API的顺序号范围，导出部分可视化数据
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        start_idx, end_idx = 1, 2
        ret, output = exec_cmd(f"hipprof --hip-trace --index-range {start_idx}:{end_idx} {exe_path}", cwd=test_env)
        assert ret is True
        expect_api = self.matrix_call_seq[start_idx-1:end_idx]  # prof内部调用api序号从1开始递增
        mylogger.debug(f"expect_api: {expect_api}")
        # 检测截取的api序列是否正确
        trace_d = self.get_hip_trace_info_from_output(output)
        assert end_idx - start_idx + 1 == trace_d["api_total_calls"] + trace_d["kernel_total_calls"]
        assert sorted(expect_api) == sorted(list(trace_d["api_stat"]))  # 终端打印的api未必按调用时间展示，故排序再判断

    @pytest.mark.high
    def test_index_range_by_db(self, test_env):
        """
        --index-range: 指定HIP API的顺序号范围，导出.db文件中的部分可视化脚本
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        # step1:运行程序生成db文件
        ret, output = exec_cmd(f"hipprof --hip-trace {exe_path}", cwd=test_env)
        files_d = self.get_all_prof_files_info(output, test_env)
        db_name = files_d["db"]["name"]

        # step2：通过步骤一生成的db导出新的json
        start_idx, end_idx = 2, 4
        ret, output = exec_cmd(f"hipprof --hip-trace --index-range {start_idx}:{end_idx} --db ./{db_name}", cwd=test_env)
        expect_api = self.matrix_call_seq[start_idx-1:end_idx]
        mylogger.debug(f"expect_api: {expect_api}")
        # 检测截取的api序列是否正确
        for line in output.splitlines():
            if "HIP PROF:hip kernel statistics" in line:
                break
            if line.strip().startswith("|hip"):
                api, calls = line.strip().replace(" ","").split("|")[1:3]  # 提取api及对应调用次数
                mylogger.debug(f"get api: {api}, calls: {calls}")
                if api not in expect_api:
                    assert False, f"get unexpected api: {api}"
                for _ in range(int(calls)):
                    expect_api.remove(api)
        assert len(expect_api) == 0

    @pytest.mark.low
    def test_invalid_index_range(self, test_env):
        """
        --index-range: 指定HIP API的顺序号范围格式错误
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --index-range 3 {exe_path}", cwd=test_env)
        assert ret is True
        # 默认生成3到最大值，不会报错
        assert self.check_trace_output_success(output) is True

    @pytest.mark.smoke
    @pytest.mark.high
    def test_group_stream(self, test_env):
        """
        --group-stream: 将可视化文件的memcpy与核函数的数据一起显示在流中
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --group-stream {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True
        # 获取result_PID_stream.json路径
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        # 检测json文件的目录结构，只显示[1]Runtime API、[2]Stream on Device {DCU_ID}
        check_api = check_stream = False
        for trace_event in self.parse_prof_json(json_path):
            if trace_event.get("ph") == "M":  # 事件分类只在json的元数据Metadata中
            # if trace_event.get("args") and len(trace_event.get("args")) == 1:
                category_name = trace_event.get("args").get("name", "")
                if not re.match(r"^\[[\d+]\]", category_name):
                    assert False, f"Obtained metadata of an unknown type: {category_name}"
                # if category_name == "[1] Runtime API":
                if category_name.startswith("[1] Runtime API"):  # dtk-25.10增添了ip等信息
                    check_api = True
                elif category_name.startswith("[2] Stream on Device "):
                    check_stream = True
                else:
                    assert False, f"unexpected category_name: '{category_name}' in trace_event:{trace_event}"
        assert check_api == check_stream == True

    def test_group_stream_by_db(self, test_env):
        """
        --group-stream: 将可视化文件的memcpy与核函数的数据一起显示在流中（根据db文件重新导出）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace {exe_path}", cwd=test_env)
        assert ret is True and self.check_trace_output_success(output) is True
        pid = self.get_hip_trace_info_from_output(output)["pid"]
        db_name = self.get_all_prof_files_info(output, test_env)["db"]["name"]
        ret, output = exec_cmd(f"hipprof --hip-trace --group-stream --db ./{db_name}", cwd=test_env)
        assert ret is True
        # assert self.check_trace_output(output, test_env) is True
        # 获取新的hip-prof-PID.db_stream.json路径
        new_json_name = f"hip-prof-{pid}.db_stream.json"
        # # 定义正则表达式模式
        # pattern = re.compile(r'\[2\] Stream on Device \d+')
        # 检测json文件的目录结构，只显示[1]Runtime API、[2]Stream on Device {DCU_ID}
        check_api = check_stream = False
        for trace_event in self.parse_prof_json(new_json_name):
            if trace_event.get("args") and len(trace_event.get("args")) == 1:
                category_name = trace_event.get("args").get("name", "")
                if not re.match(r"^\[[\d+]\]", category_name):
                    break
                if category_name.startswith("[1] Runtime API"):
                    check_api = True
                elif category_name.startswith("[2] Stream on Device "):
                    check_stream = True
                else:
                    assert False, f"unexpected category_name: '{category_name}' in trace_event:{trace_event}"
        assert check_api == check_stream == True

    @pytest.mark.high
    def test_output_type_json(self, test_env):
        """
        --output-type: 设置导出的可视化文件为json格式（--output-type参数为0）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --output-type 0 {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 检测只生成json文件，无html文件
        files_d = self.get_all_prof_files_info(output, test_env)
        assert len(files_d["json_list"]) > 0 and len(files_d["html_list"]) == 0

    @pytest.mark.high
    def test_output_type_html(self, test_env):
        """
        --output-type: 设置导出的可视化文件为json格式（--output-type参数为1）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --output-type 1 {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

        # 检测只生成json文件，无html文件
        files_d = self.get_all_prof_files_info(output, test_env)
        assert len(files_d["json_list"]) == 0 and len(files_d["html_list"]) > 0

    def test_buffer_size_min(self, test_env):
        """
        设置性能缓存数据最小为 1
        :param test_env:
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --buffer-size 1 {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_trace_output_success(output) is True

    def test_trace_memcpy_type(self, test_env):
        """
        测试跟踪不同方向的内存拷贝类型
        :param test_env:
        :return:
        """
        # 1、编译待测用例（用例依次使用内存拷贝H2D,D2D,D2H,H2H四种类型）
        exe_path = self.compile_source_code(src_file_name="memory_copy/multiple_type_memcpy.cpp")

        # 2、执行测试命令
        ret, output = exec_cmd(f"hipprof --hip-trace --trace-args {exe_path}", cwd=test_env)
        assert ret is True
        # assert self.check_trace_output_success(output) is True

        # 3、提取与检查json数据
        json_path = self.get_all_prof_files_info(output, test_env)["json_list"][0]["abspath"]
        trace_events_l = self.parse_prof_json(json_path)

        memcpy_trace_event_l = []  # 提取内存拷贝的事件
        kernel_trace_event_l = []  # 提取kernel ops
        for trace_event_d in trace_events_l:
            if trace_event_d.get("ph") == "X":
                if trace_event_d.get("name", "") in ["HostToDevice", "DeviceToDevice", "DeviceToHost"]:
                    memcpy_trace_event_l.append(trace_event_d)
                elif trace_event_d.get("name", "") == "squareElements(float*, unsigned long)":
                    kernel_trace_event_l.append(trace_event_d)
        mylogger.debug(f"memcpy_trace_event_l: {memcpy_trace_event_l}\nkernel_trace_event_l: {kernel_trace_event_l}")

        # 校验内存拷贝类型
        assert len(memcpy_trace_event_l) == 4
        for seq, memcpy_trace_event in enumerate(memcpy_trace_event_l):
            if seq == 0:
                assert memcpy_trace_event.get("name") == "HostToDevice"
                assert memcpy_trace_event.get("args").get("source memory kind") == "Pinned"
                assert memcpy_trace_event.get("args").get("destination memory kind") == "Device"
            elif seq == 1:
                assert memcpy_trace_event.get("name") == "DeviceToDevice"
                assert memcpy_trace_event.get("args").get("source memory kind") == "Device"
                assert memcpy_trace_event.get("args").get("destination memory kind") == "Device"
            elif seq == 2:
                assert memcpy_trace_event.get("name") == "DeviceToHost"
                assert memcpy_trace_event.get("args").get("source memory kind") == "Device"
                assert memcpy_trace_event.get("args").get("destination memory kind") == "Pinned"
            elif seq == 3:
                assert memcpy_trace_event.get("name") == "DeviceToHost"
                assert memcpy_trace_event.get("args").get("source memory kind") == "Pinned"
                assert memcpy_trace_event.get("args").get("destination memory kind") == "Pinned"
        # 校验核函数网格参数
        assert len(kernel_trace_event_l) == 2
        for seq, kernel_trace_event in enumerate(kernel_trace_event_l):
            kernel_args = kernel_trace_event.get("args", {}).get("args", "")
            assert re.search(r"numBlocks\s?=\s?{z=16,\s?y=4,\s?x=1}", kernel_args) is not None
            assert re.search(r"dimBlocks\s?=\s?{z=2,\s?y=4,\s?x=8}", kernel_args) is not None
            assert "numBlocks={z=16, y=4, x=1}" in kernel_trace_event.get("args", {}).get("args")
            assert "dimBlocks={z=2, y=4, x=8}" in kernel_trace_event.get("args", {}).get("args")
        # 校验核函数运行在不同卡上
        assert int(kernel_trace_event_l[0].get("args").get("dev_id") or kernel_trace_event_l[0].get("args").get("devId") ) != int(
            kernel_trace_event_l[1].get("args").get("dev_id") or kernel_trace_event_l[1].get("args").get("devId"))
        # TODO待perfto功能实现后，观察json的pushconfig和popconfig是否还有重叠


@allure.feature('trace组合指令测试')
@pytest.mark.medium
class TestCombinedParasTrace(BasePublic):
    """组合参数功能"""
    def test_hiprtc_trace(self, test_env):
        pass





